package cep;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;
import java.util.Objects;

public class TestCep01 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> stringDataStreamSource = env.readTextFile("data/cep01.txt");
        SingleOutputStreamOperator<UserRegist> ds = stringDataStreamSource.map(new MapFunction<String, UserRegist>() {
            @Override
            public UserRegist map(String value) throws Exception {
                String[] split = value.split(" ");
                return new UserRegist(Integer.valueOf(split[0]), split[1], split[2]);
            }
        });


//        KeyedStream<UserRegist, String> keybyDS = ds.keyBy(f -> f.name);

        Pattern<UserRegist, UserRegist> pattern = Pattern.<UserRegist>begin("match_regist").where(new SimpleCondition<UserRegist>() {
            @Override
            public boolean filter(UserRegist value) throws Exception {
                return value.action.equals("regist");
            }
        }).timesOrMore(5).until(new IterativeCondition<UserRegist>() {
            @Override
            public boolean filter(UserRegist userRegist, Context<UserRegist> context) throws Exception {
                Iterable<UserRegist> it = context.getEventsForPattern("match_regist");
                int i =0;
                for(UserRegist ur:it){
                    i+=1;
                }

                //判断的条件 当==5时截断 或者不等于regist时候
                if(i==5||!userRegist.action.equals("regist")){
                    return true;
                }else{
                    return false;
                }
            }
        });

        PatternStream<UserRegist> ps = CEP.pattern(ds, pattern);

        ps.select(new PatternSelectFunction<UserRegist, String>() {
            @Override
            public String select(Map<String, List<UserRegist>> map) throws Exception {
                System.out.println("11111");
                map.forEach((a,b)->{
                    System.out.println(a);
                    b.forEach(e-> System.out.println(e));
                });
                System.out.println("2222");
                List<UserRegist> match_regist = map.get("match_regist");
                if(match_regist!=null&&match_regist.size()>0)
                    return match_regist.get(match_regist.size()-1).name;
                else
                    return "no_match";
            }
        }).print();
        env.execute();
    }

    public static class UserRegist{
        private int id;
        private String name;
        private String action;

        public UserRegist() {
        }

        public UserRegist(int id, String name, String action) {
            this.id = id;
            this.name = name;
            this.action = action;
        }

        @Override
        public String toString() {
            return "UserRegist{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    ", action='" + action + '\'' +
                    '}';
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (!(o instanceof UserRegist)) return false;
            UserRegist that = (UserRegist) o;
            return getId() == that.getId() &&
                    Objects.equals(getName(), that.getName()) &&
                    Objects.equals(getAction(), that.getAction());
        }

        @Override
        public int hashCode() {

            return Objects.hash(getId(), getName(), getAction());
        }

        public int getId() {
            return id;
        }

        public void setId(int id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String getAction() {
            return action;
        }

        public void setAction(String action) {
            this.action = action;
        }
    }
}

